package iscas;

import redis.clients.jedis.Jedis;

import java.util.concurrent.atomic.AtomicReference;
import java.util.function.BiFunction;
import java.util.function.Function;

public class LocalValueState<T> extends RedisLocalState {
    AtomicReference<T> local = new AtomicReference<>();
    T defaultValue;
    BiFunction<T, T, T> merger;
    Function<String, T> parser;

    public LocalValueState(T local, T defaultValue, String name, BiFunction<T, T, T> merger,
                           Function<String, T> parser, long period) {
        super(name, period);
        this.local.set(local);
        this.defaultValue = defaultValue;
        this.merger = merger;
        this.parser = parser;
    }

    public void set(T v) {
        local.set(v);
    }

    public T get() {
        return local.get();
    }

    @Override
    boolean isEmpty() {
        return false;
    }

    @Override
    public void merge(Jedis jedis) {
        lock(jedis);
        String remoteString = jedis.get(name);
        T remote = defaultValue;
        if (remoteString != null) {
            remote = parser.apply(remoteString);
        }
        T merged = merger.apply(local.getAndSet(defaultValue), remote);
        jedis.set(name, merged.toString());
        unlock(jedis);
    }
}
